package com.lxj.utils;



import com.lxj.bean.TransientSink;
import com.lxj.common.GmallConfig;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/*
*
jdbc sink   写死谁都会
*
* 写活
*
*
*
* */
public class MyClickHouseUtil {

    //T  代表输入类型还是输出类型
    //T  输入类型
    public static <T> SinkFunction<T> getSinkFunction(String sql) {



        return JdbcSink.<T>sink(sql,
                /*
                * 通过反射获取属性名称
                * 通过属性名称获取对应的value
                * 通过属性获取注解  判断注解是否存在
                * */
//                insert into dws_traffic_source_keyword_page_view_window  values(?,?,?,?,?,?)
                new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        //使用反射的方式获取t对象中的数据
                        Class<?> tClz = t.getClass();

                        //获取并遍历属性
                        Field[] declaredFields = tClz.getDeclaredFields();



                        int offset = 0;   //1
                        for (int i = 0; i < declaredFields.length; i++) {

                            //获取单个属性
                            Field field = declaredFields[i];
                            //暴力访问
                            field.setAccessible(true);

                            //尝试获取字段上的自定义注解
                            TransientSink transientSink = field.getAnnotation(TransientSink.class);


                            if (transientSink != null) {
                                offset++;
                                continue;
                            }

                            //获取属性值  //属性名称  属性值
                            Object value = null;
                            try {
                                value = field.get(t);
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                            }

                            //给占位符赋值
                            preparedStatement.setObject(i + 1 - offset, value);

                        }
                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(1000L)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build());
    }

}
